package com.wo56.pushmsgcenter.common.exe;

import java.util.Map;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.framework.components.redis.RedisHelper;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.ShardedJedis;

public class TestRedisBrpop {

	public static void main(String[] args) {
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout("spout", new RedisSpout(), 4);
		builder.setBolt("sysout", new SplitSentence(), 4).shuffleGrouping("spout");
		Config config = new Config();
		config.setDebug(true);
		// 这里是本地模式下运行的启动代码。
		config.setMaxTaskParallelism(1);
		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology("simple", config, builder.createTopology());
	}

	
	public static class RedisSpout extends BaseRichSpout {
		SpoutOutputCollector _collector;
		@Override
		public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
			// TODO Auto-generated method stub
			_collector = collector;
		}

		@Override
		public void nextTuple() {
			// TODO Auto-generated method stub
			Jedis jedis = null;
			if(RedisHelper.getInstance().isShared()){
				ShardedJedis shardedJedis = (ShardedJedis) RedisHelper.getInstance().getJedis();
			}else{
				jedis = (Jedis) RedisHelper.getInstance().getJedis();
			}
			while(true){
				_collector.emit(new Values(jedis.brpop(0, "popTest").get(1)));
//				System.out.println(jedis.brpop(0, "popTest").get(1));
			}
		}

		@Override
		public void declareOutputFields(OutputFieldsDeclarer declarer) {
			// TODO Auto-generated method stub
			declarer.declare(new Fields("num"));
		}
		
	}
	
	
	public static class SplitSentence extends BaseBasicBolt {
	    @Override
	    public void execute(Tuple tuple, BasicOutputCollector collector) {
	      String sentence = tuple.getString(0);
	      System.out.println(sentence);
	    }

	    @Override
	    public void declareOutputFields(OutputFieldsDeclarer declarer) {
	    }
	  }

	
}
